package org.apache.spark.streaming

import java.util.Properties

import com.aliyun.openservices.ons.api._

object AliOnsReceiver {
  def main(args: Array[String]): Unit = {
    val consumerId = "CID-bigdata";
    val accessKey = "arzNQ1bmmyOSjHtt";
    val secretKey = "yFHjBg4HXVwBSyMkTrxQWcuinAi2o7";
    val topic = "bigdata";
    val key = "*";

    val properties: Properties = new Properties
    properties.put(PropertyKeyConst.ConsumerId, consumerId)
    properties.put(PropertyKeyConst.AccessKey, accessKey)
    properties.put(PropertyKeyConst.SecretKey, secretKey)
    val consumer: Consumer = ONSFactory.createConsumer(properties)

    consumer.subscribe(topic, key, new MessageListener() {
      def consume(message: Message, context: ConsumeContext): Action = {
//        System.out.println("Receive=1=: " + message)

        var as = new String(message.getBody,"utf-8");
        System.out.println("Receive=2=: " + as)
        return Action.CommitMessage
      }
    })
    consumer.start
    System.out.println("Consumer Started")
  }
}
